package com.wxl.common.util

import io.reactivex.Observable
import io.reactivex.ObservableOnSubscribe
import io.reactivex.Observer
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.Disposable
import io.reactivex.functions.Consumer
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnit

class ObservableUtil {
    companion object {


       open abstract class OnNextCallback<T>{
           abstract fun onNext(t:T)
          open fun error(e: Throwable){}
        }


        fun delay(time:Long,r: Runnable){
            Observable.timer(time,TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.newThread())
                .subscribe(object : Observer<Long>{
                    override fun onComplete() {

                    }

                    override fun onSubscribe(d: Disposable) {

                    }

                    override fun onNext(t: Long) {
                        r.run()
                    }

                    override fun onError(e: Throwable) {

                    }

                })
        }

        fun <T> doNext(oos:ObservableOnSubscribe<T>,next:OnNextCallback<T>){
            Observable.create(oos)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.newThread())
                .subscribe(object : Observer<T>{
                    override fun onComplete() {

                    }

                    override fun onSubscribe(d: Disposable) {

                    }

                    override fun onNext(t: T) {
                        next.onNext(t)
                    }

                    override fun onError(e: Throwable) {
                        next.error(e)
                    }

                })

        }

        fun tick(delay:Long,period:Long,take:Long,onNext: Consumer<Long>,onDisposable:Consumer<Disposable>?){
            Observable.interval(delay,period,TimeUnit.MILLISECONDS)
                .take(take)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(object :Observer<Long>{
                    override fun onComplete() {

                    }

                    override fun onSubscribe(d: Disposable) {
                        onDisposable?.accept(d)
                    }

                    override fun onNext(t: Long) {
                        onNext.accept(t)
                    }

                    override fun onError(e: Throwable) {
                    }

                })
        }


    }
}